package mn.idax.exchange.otc.service.impl;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import mn.idax.exchange.auth.bean.AuthInfo;
import mn.idax.exchange.auth.interfaces.AuthService;
import mn.idax.exchange.auth.util.AuthHelper;
import mn.idax.exchange.common.SpringUtils;
import mn.idax.exchange.otc.constant.CommonConstant;
import mn.idax.exchange.otc.constant.RequestUrlConstant;
import mn.idax.exchange.otc.domain.ConnectionDTO;
import mn.idax.exchange.otc.domain.MessageNotifyDTO;
import mn.idax.exchange.otc.domain.MongoMessageNotifyDTO;
import mn.idax.exchange.otc.domain.PushNotifyMessage;
import mn.idax.exchange.otc.service.MongoDBService;
import mn.idax.exchange.otc.util.LogUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.atmosphere.config.service.Disconnect;
import org.atmosphere.config.service.ManagedService;
import org.atmosphere.config.service.Message;
import org.atmosphere.config.service.Ready;
import org.atmosphere.cpr.AtmosphereResource;
import org.atmosphere.cpr.AtmosphereResourceEvent;
import org.atmosphere.cpr.Broadcaster;
import org.atmosphere.cpr.BroadcasterFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.inject.Inject;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Auther: dison
 * @Date: 2018/8/31 13:00
 * @Description:
 *  参考连接：https://blog.csdn.net/cyuanxin/article/details/52751712
 */
@ManagedService(path = RequestUrlConstant.NOTIFY_REQUEST_PATH)
@Slf4j
@Service
public class MessageNotifyService {

    public static final ConcurrentHashMap<String, ConnectionDTO> CONNECTIONS = new ConcurrentHashMap<>();
    /**
     * 用来标识用户是否传了token
     */
    private static final String NOTIFYATTRIBUTES = "notifyAttributes";

    @Inject
    public static BroadcasterFactory broadcasterFactory;
    @Inject
    private static Broadcaster systemBroadcaster; // 系统广播通知


    /**
     *
     * 功能描述:
     *      建立连接
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:40
     */
    @Ready
    public void onReady(AtmosphereResource resource){

        try {
            String token = resource.getRequest().getPathInfo();

            ConnectionDTO connectionDTO = new ConnectionDTO();
            connectionDTO.setTimestamp(System.currentTimeMillis());
            //新的ws连接
            if(StringUtils.isEmpty(token) || token.length()<2){
                //将该连接信息丢到内存里
                connectionDTO.setResource(resource);

                CONNECTIONS.put(resource.uuid(),connectionDTO);
            } else {
                token = token.substring(1);
                if (!this.authentication(token)){
                    this.closeResource(resource);
                    return;
                }
                AuthInfo authInfo = getAuthInfo(token);
                connectionDTO.setUserId(authInfo.getId());

                Broadcaster broadcaster = broadcasterFactory.lookup(authInfo.getId(), true);
                //add AtmosphereResource to broadcaster
                broadcaster.addAtmosphereResource(resource);
            }

            // 将所有登录的用户加入到系统广播组中
            this.getSystemBroadcasterInstance().addAtmosphereResource(resource);
            resource.getRequest().getSession().setAttribute(NOTIFYATTRIBUTES,connectionDTO);

        } catch (Exception e) {
            try {
                resource.close();
            } catch (IOException ex) {
                LogUtils.warn(this.getClass(),"onReadyIO fail  " + ExceptionUtils.getStackTrace(ex));
            }
            LogUtils.warn(this.getClass(),"messageNotifyServiceOnReady fail  " + ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     *
     * 功能描述:
     *      关闭连接
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:40
     */
    @Disconnect
    public void onDisconnect(AtmosphereResourceEvent event) {
        log.info("Client {} disconnected [{}]", event.getResource().uuid(),
                event.isCancelled() ? "cancelled" : "closed");
    }

//    /**
//     *
//     * 功能描述:
//     *      消息监听
//     * @param:
//     * @return:
//     * @auther: dison
//     * @date: 2018/8/31 13:40
//     */
//    @org.atmosphere.config.service.Message(encoders = JacksonEncoderDecoder.class, decoders = JacksonEncoderDecoder.class)
//    public MessageNotifyDTO onMessage(AtmosphereResource resource, MessageNotifyDTO message) throws IOException {
//        log.info("notifyMessage receive message {}", JSON.toJSON(message));
//
//        return message;
//    }

    @Message
    public void onMessage(AtmosphereResource resource, String message) {

        try {
            LogUtils.info(MessageNotifyService.class,"notifyWS onMessage receive data" + message);
            ConnectionDTO attribute = (ConnectionDTO)resource.getRequest().getSession(false).getAttribute(NOTIFYATTRIBUTES);
            if (Objects.isNull(attribute.getUserId())) {
                if (StringUtils.isEmpty(message)){
                    return ;
                }

                Map<String,String> map = JSON.parseObject(message, Map.class);
                String token = map.get("token");
                if (!this.authentication(token)){
                    this.closeResource(resource);
                }
                AuthInfo authInfo = getAuthInfo(token);
                attribute.setUserId(authInfo.getId());

                resource.getRequest().getSession().setAttribute(NOTIFYATTRIBUTES, attribute);

                Broadcaster broadcaster = broadcasterFactory.lookup(authInfo.getId(), true);
                //add AtmosphereResource to broadcaster
                broadcaster.addAtmosphereResource(resource);

                ConnectionDTO connectionDTO = CONNECTIONS.get(resource.uuid());
                connectionDTO.setTimestamp(0);
                LogUtils.info(this.getClass(),"notifyWSOnMessageResource uuid is " + resource.uuid());
            }
        } catch (Exception e) {
            try {
                resource.close();
            } catch (IOException ex) {
                LogUtils.warn(this.getClass(),"onMessageIO fail  " + ExceptionUtils.getStackTrace(ex));
            }
            LogUtils.warn(this.getClass(),"messageNotifyServiceOnMessage fail  " + ExceptionUtils.getStackTrace(e));
        }
    }

    public Broadcaster getSystemBroadcasterInstance(){
        if (systemBroadcaster == null){
            synchronized (MessageNotifyService.class){
                if (systemBroadcaster == null){
                    systemBroadcaster = broadcasterFactory.lookup(RequestUrlConstant.NOTIFY_SYSTEM_NAME, true);
                }
            }
        }
        return systemBroadcaster;
    }

    /**
     *
     * 功能描述:
     *      关闭连接
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:33
     */
    private void closeResource(AtmosphereResource resource) {
        try {
            resource.close();
        } catch (IOException e) {
            log.error("----close connect error, connect resource uuid:{}----", resource.uuid());
        }
    }

    /**
     *
     * 功能描述:
     *      用户登录鉴权
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:28
     */
    public boolean authentication(String token){
        AuthService authService = SpringUtils.getBean(AuthService.class);
        boolean isTokenValid = AuthHelper.getInstance(authService).isTokenValid(token);
        if (isTokenValid){
            return true;
        }
        return false;
    }

    /**
     *
     * 功能描述:
     *      获取用户登录信息
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:29
     */
    public AuthInfo getAuthInfo(String token){
        AuthService authService = SpringUtils.getBean(AuthService.class);
        AuthInfo authInfo = AuthHelper.getInstance(authService).decodeToken(token);
        return authInfo;
    }

    /**
     *
     * 功能描述:
     *      发送系统消息
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:29
     */
    public void broadcastToSystem(MessageNotifyDTO message) {
        this.getSystemBroadcasterInstance().broadcast(message);
    }

    /**
     *
     * 功能描述:
     *      发送个人消息
     * @param:
     * @return:
     * @auther: dison
     * @date: 2018/8/31 13:32
     */
    public boolean broadcastToUser(Integer userId,MessageNotifyDTO message) {
        MongoMessageNotifyDTO mongoMessage = new MongoMessageNotifyDTO();
        BeanUtils.copyProperties(message,mongoMessage);
        mongoMessage.setReceiver(userId);
        mongoMessage.setReadFlag(0);
        MongoDBService mongoDBService = SpringUtils.getBean(MongoDBService.class);
        mongoDBService.insert(mongoMessage);

        Broadcaster broadcaster = broadcasterFactory.lookup(userId, true);
        if (broadcaster.isDestroyed()){
            LogUtils.info(this.getClass() , "broadcastToUserFail,broadcaster is destroyed｛" + userId + "｝未连接导致无法推送消息");
            return false;
        }
        PushNotifyMessage pushNotifyMessage = PushNotifyMessage.builder()
                                                            .message(JSON.toJSONString(message))
                                                            .userId(userId)
                                                            .build();

        this.pushRedisMessage(CommonConstant.PUSH_NOTIFY_TOPIC , JSON.toJSONString(pushNotifyMessage));
        return true;
    }

    public void pushRedisMessage(String topicName ,String message){
        RedisService redisService = SpringUtils.getBean(RedisService.class);

        redisService.pushMessage(topicName , message);
    }
}
